// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: leon

#pragma once

#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>

template <typename T>
class ConcurrentQueue {
 public:
  T Pop() {
    std::unique_lock<std::mutex> lock(mutex_);
    while (queue_.empty()) {
      cond_.wait(lock);
    }
    auto val = queue_.front();
    queue_.pop();
    lock.unlock();
    cond_.notify_one();
    return val;
  }

  void Push(const T& item) {
    std::unique_lock<std::mutex> lock(mutex_);
    while (queue_.size() >= queue_sizes_) {
       cond_.wait(lock);
    }
    queue_.push(item);
    lock.unlock();
    cond_.notify_one();
  }

  bool IsEmpty() {
    std::unique_lock<std::mutex> lock(mutex_);
    return queue_.empty();
  }

  void NotifyAll() {
    cond_.notify_all();
  }

  explicit ConcurrentQueue(int32_t size) : queue_sizes_(size) {}
  ConcurrentQueue(const ConcurrentQueue&) = delete;
  ConcurrentQueue& operator=(const ConcurrentQueue&) = delete;

 private:
  std::queue<T> queue_;
  std::mutex mutex_;
  std::condition_variable cond_;
  int32_t queue_sizes_;
};
